package com.niit.covid.sort;


import com.niit.covid.bean.CovidCountBean2;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

/**
 * @author:Ys
 * @date: 2022年08月11日 14:46
 * @desc:
 *  需求：将每个州累计确诊病例，每个州state的确诊案例数进行倒序排序。
 * 分析：
 *  如果你的需求中需要根据某个属性进行排序 ，不妨把这个属性作为key。因为MapReduce中key有默认排序行为的。但是需要进行如下考虑：
 *  如果你的需求是正序，并且数据类型是Hadoop封装好的基本类型。这种情况下不需要任何修改，直接使用基本类型作为key即可。因为Hadoop封装好的类型已经实现了排序规则。
 */
public class CovidSortSumMapper extends Mapper<LongWritable, Text, CovidCountBean2, Text> {

    /**
     * 3、创建输出对象
     */
    CovidCountBean2 outKey = new CovidCountBean2();
    Text outValue = new Text();


    /**
     * 1、重写map父类方法：map回车
     */

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        //Alabama	确诊病例数：319025	死亡病例数：11654
        /**
         * 2、读取一行数据进行切割
         */
        String[] split = value.toString().split("\t");
        //  split === [Alabama,确诊病例数：319025,死亡病例数：11654]
        //                0          1             2
        String[] casesData = split[1].split("：");//split[1] 是一个字符串 “确诊病例数：319025”
        //split[1]     ===》 确诊病例数：319025
        //[确诊病例数,319025]
        String[] deathsData = split[2].split("：");
        //split[2]    ==>  死亡病例数：11654
        //[死亡病例数,11654]
        /**
         * 4、提取数据 州、确诊数、死亡数
         */
        String state = split[0];
        long cases = Long.parseLong(casesData[1]) ;
        long deaths = Long.parseLong(deathsData[1]);
        /**
         * 5、输出结果
         */
       outKey.set(cases,deaths);
       outValue.set(state);

       context.write(outKey,outValue);

    }

}
